package com.nc.mem.serial.graph.asm;

import java.io.InputStream;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.objectweb.asm.Type;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.nc.mem.GraphClassLoader;
import com.nc.mem.PrivateAccessor;
import com.nc.mem.serial.Serializer;
import com.nc.mem.serial.graph.Context;
import com.nc.mem.serial.graph.GraphSerializer;
import com.nc.mem.serial.graph.Representation;
import com.nc.mem.serial.graph.common.CompatibleSerializer;
import com.nc.mem.serial.graph.container.SerializerRegistry;
import com.nc.mem.serial.graph.cust.CollectionHolder;
import com.nc.mem.serial.graph.cust.FixedKeyValMapSerializer;
import com.nc.mem.serial.graph.cust.FixedTypeCollectionSerializer;
import com.nc.mem.serial.graph.cust.MapHolder;
import com.nc.mem.serial.graph.cust.NotNullNonPolymorphic;
import com.nc.mem.serial.graph.cust.Serializable;

public final class GraphSerializerFactory {

	public static boolean TRACE = false;
	public static boolean SAVE = true;

	static final SerializerRegistry REPO;

	static {
		REPO = SerializerRegistry.getInstance();
	}

	public static Serializer alternativeOrDefault(final Class<?> clazz) {
		Serializer ser = REPO.getAlternative(clazz);

		if (ser == null) {
			if (Map.class.isAssignableFrom(clazz)) {
				ser = FixedKeyValMapSerializer.INSTANCE;
				REPO.register(clazz, ser);
			} else if (Collection.class.isAssignableFrom(clazz)) {
				ser = FixedTypeCollectionSerializer.INSTANCE;
				REPO.register(clazz, ser);
			}
		}

		if (ser == null) {
			if (REPO.isForCompatibility(clazz)) {
				ser = CompatibleSerializer.INSTANCE;
			} else {
				ser = REPO.serializer(clazz);
				if (ser == null) {
					/*
					 * Here comes a hacky part: if there's a cycle between final classes, say A and
					 * B, A will try to get the serializer of B, whose class may or may not have
					 * been created. If it doesn't exist it will fail, then we'll have to create it.
					 * Upon initialization of class B, it will try to reentrantly resolve A, which
					 * has it's field "INSTANCE" already defined, so B will succeed. Once the
					 * 'recursive' call returns, the catch block will succeed because B will be
					 * cached already. This is more or less how the VM resolves 2 cyclic singletons.
					 */
					try {
						ser = (Serializer) Class.forName(clazz.getName() + "$GRAPH_SER").getDeclaredField("INSTANCE").get(null);
					} catch (final Throwable e) {
						ser = serializer(clazz);
					}
				}
			}
		}

		if (ser == null) {
			throw new RuntimeException("Failed to created alternative serializer!");
		}

		return ser;
	}

	public static Serializer createByIntrospection(final Class<?> clazz, final boolean isFullGraph, final boolean register) {
		Serializer serializer;

		final boolean fullGraph = isFullGraph || clazz.getName().startsWith("java");
		final Serializable anno = clazz.getAnnotation(Serializable.class);
		final boolean includeAll = fullGraph || REPO.isClosure(clazz) || anno != null && anno.include().length == 1 && "*".equals(anno.include()[0]);

		final Set<String> include = includeAll ? Collections.<String> emptySet() : anno == null ? Collections.<String> emptySet() : new HashSet<String>(Arrays.asList(anno.include()));
		final Set<String> exclude = includeAll ? Collections.<String> emptySet() : anno == null ? Collections.<String> emptySet() : new HashSet<String>(Arrays.asList(anno.exclude()));

		final boolean filterInclude = include != null && !include.isEmpty();
		final boolean filterExclude = !filterInclude && exclude != null && !exclude.isEmpty();

		boolean serPvt = false || fullGraph || REPO.isClosure(clazz);
		final Class<?> root = clazz;
		final String rootInternalName = Type.getInternalName(clazz);
		final String owner = rootInternalName;
		final List<FieldInfo> prims = new ArrayList<FieldInfo>();
		final List<FieldInfo> refs = new ArrayList<FieldInfo>();
		final List<FieldInfo> notPolyRefs = new ArrayList<FieldInfo>();
		final Field[] fields = clazz.getDeclaredFields();

		final PrivateAccessor a = PrivateAccessor.INSTANCE;

		for (final Field field : fields) {
			final int mods = field.getModifiers();

			// bypass: ignore static and synthetic fields and honor transient semantics
			if (Modifier.isStatic(mods) || Modifier.isTransient(mods)) {
				continue;
			}

			// Final fields must be handled as private upon deserialization. We can still use direct
			// access on serialization.
			if (field.isSynthetic()) {
				if (REPO.isClosure(clazz)) {
					serPvt = true;
				} else {
					continue;
				}

			} else if (Modifier.isPrivate(mods) || Modifier.isFinal(mods)) {
				if (a == null || !includeAll && !include.contains(field.getName())) {
					continue;
				}

				serPvt = true;
			}

			if (!includeAll && (filterInclude && !include.contains(field.getName()) || //
					filterExclude && exclude.contains(field.getName()))) {
				continue;
			}

			final long offset = a == null ? -1l : a.objectFieldOffset(field);

			final ReflexiveFieldInfo info = new ReflexiveFieldInfo(owner, field, offset, field.getModifiers());
			if (field.getType().isPrimitive()) {
				prims.add(info);
			} else {
				refs.add(info);
				if (isNotPolymorphic(field)) {
					notPolyRefs.add(info);
				}
			}
		}

		Class<?> sc = clazz.getSuperclass();
		Class<?> ipwf = null;

		outer: while (sc != Object.class) {
			final Field[] scFields = sc.getDeclaredFields();

			for (final Field f : scFields) {

				final int mods = f.getModifiers();

				// bypass
				if (Modifier.isStatic(mods) || Modifier.isTransient(mods) || f.isSynthetic()) {
					continue;
				}

				if (isFullGraph || REPO.isFullGraph(sc) || !Modifier.isPrivate(mods)) {
					ipwf = sc;
					break outer;
				}
				// found a serializable field, that is, there is a corresponding
				// serializer to be constructed

			}

			sc = sc.getSuperclass();
		}

		Collections.sort(prims);
		Collections.sort(refs);
		Collections.sort(notPolyRefs);

		InputStream stream = null;

		try {
			stream = GraphSerializer.class.getResourceAsStream(GraphSerializer.class.getSimpleName() + ".class");
			final GraphClassAdapter adapter = new GraphClassAdapter(clazz);
			adapter.ipwf = ipwf;
			adapter.prims = prims;
			adapter.refs = refs;
			adapter.notPolyRefs = notPolyRefs;
			adapter.serializePrivate = serPvt;
			ClassLoader cl = clazz.getClassLoader();

			try {
				if (cl == null) {
					getLogger().warn("Attempting to create serializer for class without class loader!");
					adapter.setAccessAsPrivate();
					adapter.serializePrivate = true;
					cl = GraphClassLoader.LOADER;
				}

				serializer = adapter.create(cl, SAVE, TRACE ? System.out : null);
			} catch (final Throwable err) {
				if (err instanceof NoClassDefFoundError || err instanceof ClassNotFoundException) {
					getLogger().warn("Unable to define class using the same ClassLoader of {}. Most likely, this class is on a higher CL. We will disable bytecode field access and attempt to create a serializer using direct access on this classloader.", root.getClass().getSimpleName());

					adapter.setAccessAsPrivate();
					adapter.serializePrivate = true;
					serializer = adapter.create(cl, SAVE, TRACE ? System.out : null);
				} else {
					throw err;
				}
			}

			if (register) {
				REPO.register(root, serializer);
			}

			return serializer;
		} catch (final Throwable exc) {
			throw new RuntimeException("Unable to create serializer for " + clazz, exc);
		} finally {
			if (stream != null) {
				try {
					stream.close();
				} catch (final Exception ignore) {
					ignore.printStackTrace();
				}
			}
		}
	}

	private static Serializer createFor(final Class<?> clazz) {
		Serializer serializer = null;
		synchronized (clazz) {
			serializer = REPO.serializer(clazz);

			if (serializer == null) {
				if (clazz.isArray()) {
					// return new ArraySerializer(clazz);
					throw new UnsupportedOperationException();
				}

				if (REPO.isClosure(clazz)) {
					serializer = CompatibleSerializer.INSTANCE;
					REPO.register(clazz, serializer);
				} else if (!REPO.isFullGraph(clazz)) {
					Class<?> c = clazz;

					while ((c = c.getSuperclass()) != null && c != Object.class) {
						if (REPO.isAbstracted(c)) {
							serializer = REPO.serializer(c);
							REPO.register(clazz, serializer);
							break;
						}
					}
				}

				// attempt to register to notify collection listeners
				REPO.registerClass(clazz);

				serializer = REPO.serializer(clazz);

				if (serializer == null) {
					createByIntrospection(clazz, false, true);
					serializer = REPO.serializer(clazz);
				}
			}
		}

		return serializer;
	}

	public static void deregisterForFullGraph(final Class<?>... classes) {
		REPO.deregisterForFullGraph(classes);
	}

	static Logger getLogger() {
		return LoggerFactory.getLogger(GraphSerializerFactory.class);
	}

	static boolean isNotPolymorphic(final Field info) {
		final Class<?> type = info.getType();

		return notTrivial(type) && (Modifier.isFinal(type.getModifiers()) || info.isAnnotationPresent(NotNullNonPolymorphic.class));

	}

	private static boolean notTrivial(final Class<?> type) {
		return //
		!type.isEnum() && //
				type != Boolean.class && type != Byte.class && //
				type != Short.class && type != Character.class && //
				type != Integer.class && type != Float.class && //
				type != Long.class && type != Double.class;
	}

	/**
	 * Specially for loading a parent serializer because there might be issues in compatibility
	 * mode!
	 * 
	 * @param parent
	 * @return
	 */
	public static synchronized Serializer parentSerializer(final Class<?> parent) {
		Serializer ser = REPO.serializer(parent);

		if (ser == null) {
			ser = REPO.getAlternative(parent);

			if (ser == null) {
				try {
					ser = (Serializer) Class.forName(parent.getName() + "$GRAPH_SER").getDeclaredField("INSTANCE").get(null);
				} catch (final Throwable e) {
					ser = createByIntrospection(parent, false, !REPO.isForCompatibility(parent));
				}
			}
		}

		return ser;

	}

	public static Object readNested(final ByteBuffer src, final GraphSerializer ser, final int mask, final int idx) {
		if ((mask & 1 << idx) != 0) {
			return ser.readNonRoot(src);
		}
		return null;
	}

	/**
	 * Reads a (possibly polymorphic) reference from the stream if not null. The serializer to be
	 * used is discovered at run time by reading the class id from the stream.
	 * 
	 * @param dest
	 * @param nested
	 * @param mask
	 * @param idx
	 */
	public static Object readNested(final ByteBuffer src, final int mask, final int idx) {
		if ((mask & 1 << idx) != 0) {
			final int s = Representation.readInt(src);
			// extra guard...there will be no class if s is a pointer or null
			if (s == 1) {
				return null;
			}
			if ((s & 1 << 1) != 0) {
				return Context.current().fromId(s);
			}
			final Serializer ser = serializer(Representation.idToClass(src));
			return ser.read(src, s);

		}
		return null;
	}

	public static Serializer serializer(final Class<? extends Object> clazz) {
		Serializer serializer = REPO.serializer(clazz);
		if (serializer == null) {
			serializer = createFor(clazz);
		}
		return serializer;
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static void writeFixCol(final ByteBuffer dest, final GraphSerializer ser, final Object nested, final int mask, final int idx) {
		if ((mask & 1 << idx) != 0) {
			ser.writeNonRoot(dest, new CollectionHolder((Collection<Object>) nested, false));
		}
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static void writeFixColPayload(final ByteBuffer dest, final GraphSerializer ser, final Object nested, final int mask, final int idx) {
		if ((mask & 1 << idx) != 0) {
			ser.writeNonRoot(dest, new CollectionHolder((Collection<Object>) nested, true));
		}
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	public static void writeFixMap(final ByteBuffer dest, final GraphSerializer ser, final Object nested, final int mask, final int idx, final boolean keyOnlyPayload, final boolean valOnlyPayload) {
		if ((mask & 1 << idx) != 0) {
			ser.writeNonRoot(dest, new MapHolder((Map<Object, Object>) nested, keyOnlyPayload, valOnlyPayload));
		}
	}

	/**
	 * Writes a field of an object by knowing the serializer in advance. This method will be invoked
	 * if the nested object's class is final or is annotated with {@link NotNullNonPolymorphic}
	 * 
	 * @param dest
	 * @param ser
	 * @param nested
	 * @param mask
	 * @param idx
	 */
	public static void writeNested(final ByteBuffer dest, final GraphSerializer ser, final Object nested, final int mask, final int idx) {
		if ((mask & 1 << idx) != 0) {
			ser.writeNonRoot(dest, nested);
		}
	}

	/**
	 * Writes a (possibly polymorphic) reference to the stream if not null. The serializer to be
	 * used is discovered at run time by the nested object's class.
	 * 
	 * @param dest
	 * @param nested
	 * @param mask
	 * @param idx
	 */
	public static void writeNested(final ByteBuffer dest, final Object nested, final int mask, final int idx) {
		if ((mask & 1 << idx) != 0) {
			final Serializer child = serializer(nested.getClass());
			child.write(dest, nested);
		}
	}

}